package com.atbeijing.D03;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class MyTest1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .map(new MapFunction<SensorReading, Tuple3<String,Double,Long>>() {
                    @Override
                    public Tuple3<String,Double,Long> map(SensorReading value) throws Exception {
                        return Tuple3.of(value.id,value.temperature,1L);
                    }
                })
                .keyBy(r -> r.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<Tuple3<String, Double, Long>>() {
                    @Override
                    public Tuple3<String, Double, Long> reduce(Tuple3<String, Double, Long> value1, Tuple3<String, Double, Long> value2) throws Exception {
                        return Tuple3.of(value1.f0,value1.f1+value2.f1,value1.f2+value2.f2);
                    }
                })
                .map(new MapFunction<Tuple3<String, Double, Long>, Tuple2<String,Double>>() {
                    @Override
                    public Tuple2<String, Double> map(Tuple3<String, Double, Long> value) throws Exception {
                        return Tuple2.of(value.f0,value.f1/value.f2);
                    }
                })
                .print();
        env.execute();
    }
}
